package com.xxd.dfs.backupnode;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-08-26 10:55
 * @description: 执行checkpoint操作
 */
public class StandbyCheckpointer {
    private final Logger logger = LoggerFactory.getLogger(StandbyCheckpointer.class);
    private Configuration configuration = Configuration.getInstance();
    private String FSIMAGE_DIR = configuration.get("fsimage.dir");
    private String lastFSImageFile = "";

    /**
     * 周期检查checkpoint
     */
    private long CHECKPOINT_INTERVAL = Long.parseLong(configuration.get("checkpoint.interval"));
    /**
     * checkpoint周期执行一次 10分钟
     */
    private long CHECKPOINT_PERIOD = Long.parseLong(configuration.get("checkpoint.period"));
    private long lastCheckPointTxid = 0L;
    private long lastCheckpointTime = System.currentTimeMillis();

    private FSNameSystem namesystem;
    private NameNodeRpcClient namenode;
    ExecutorService senderFsimageThreadPool = Executors.newSingleThreadExecutor();

    public StandbyCheckpointer(FSNameSystem namesystem,
                               NameNodeRpcClient namenode,
                               String lastFSImageFile,
                               long lastCheckPointTxid) {
        this.namesystem = namesystem;
        this.namenode = namenode;
        this.lastFSImageFile = lastFSImageFile;
        this.lastCheckPointTxid = lastCheckPointTxid;
    }

    /**
     * 开启checkpoint
     */
    public void start() {
        new CheckpointerThread().start();
    }

    /**
     * 执行checkpoint 操作
     * 判断是否需要执行checkpoint
     * 1）如果超过100条edit log 没有合并到fsimage里面去
     * 2）距离上次checkpoint已经超过10分钟
     */
    private void doWork() throws IOException {
        while (true) {
            boolean needCheckpoint = false;
            try {
                Thread.sleep(CHECKPOINT_INTERVAL);
            } catch (Exception e) {
                logger.info("error", e);
            }

            long txnCount = namesystem.getTxnCount();
            if (txnCount == lastCheckPointTxid) {
                continue;
            }

            long curTime = now();
            if (curTime - lastCheckpointTime > CHECKPOINT_PERIOD) {
                logger.info("距离上次checkpoint 已经过去了 20s了, curTime = {} lastCheckpointTime = {}",
                        curTime, lastCheckpointTime);
                lastCheckpointTime = curTime;
                needCheckpoint = true;
            }

            if (txnCount - lastCheckPointTxid > 10000) {
                logger.info("txCount = {} lastCheckPointTxid = {}", txnCount, lastCheckPointTxid);
                needCheckpoint = true;
            }

            if (needCheckpoint) {
                doCheckpoint();
            }
        }
    }

    /**
     * 执行checkpoint操作
     * 1 删除上个fsimage 文件
     * 2 合成新的fsimage
     * 3 把fsimage发送给namenode
     */
    private void doCheckpoint() throws IOException {
        removeLastFSImage();
        FSImage fsimage = namesystem.getFSImage();
        writeFSImage2Disk(fsimage);
        sendCheckpointTxid(fsimage.getCheckpointTxid());
        sendFSImage2NameNode(fsimage);
        lastCheckPointTxid = fsimage.getCheckpointTxid();
    }

    /**
     * 发送checkpoint Txid 给namenode
     *
     * @param maxTxid
     */
    private void sendCheckpointTxid(long maxTxid) {
        namenode.updateCheckpointTxid(maxTxid);
    }

    /**
     * 发送FSImage 给 NameNode
     *
     * @param fsimage
     */
    private void sendFSImage2NameNode(FSImage fsimage) {
        senderFsimageThreadPool.submit(new Runnable() {
            @Override
            public void run() {
                TransferFsImage.sendFSImage2NameNode(fsimage);
            }
        });
    }

    /**
     * 讲fsimage 写入本地磁盘
     *
     * @param fsimage
     */
    private void writeFSImage2Disk(FSImage fsimage) throws IOException {
        String filePath = FSIMAGE_DIR + "\\" + "fsimage_" + fsimage.getCheckpointTxid() + ".meta";
        lastFSImageFile = filePath;

        // 本次写入的json数据
        String fsimageJson = fsimage.getFsimageJson();
        logger.info("本次写入的json数据 txid = {} filePath = {}", fsimage.getCheckpointTxid(), filePath);
        ByteBuffer buffer = ByteBuffer.wrap(fsimageJson.getBytes());

        RandomAccessFile file = null;
        FileOutputStream out = null;
        FileChannel channel = null;

        try {
            file = new RandomAccessFile(filePath, "rw");
            out = new FileOutputStream(file.getFD());
            channel = out.getChannel();

            channel.write(buffer);
            channel.force(false);
        } catch (IOException e) {
            logger.error("error", e);
        } finally {
            if (channel != null) {
                channel.close();
            }
            if (out != null) {
                out.close();
            }
            if (file != null) {
                file.close();
            }
        }
    }

    private void removeLastFSImage() {
        File file = new File(lastFSImageFile);
        if (file.exists()) {
            logger.info("正在删除上一次的 fsimage 文件 file name = {}", lastFSImageFile);
            file.delete();
        }
    }

    private long now() {
        return System.currentTimeMillis();
    }

    private class CheckpointerThread extends Thread {
        @Override
        public void run() {
            try {
                doWork();
            } catch (Exception e) {
                logger.error("error", e);
            }
        }
    }
}
